package dbtask

import (
	daogext "github.com/darwinOrg/daog-ext"
	dgcoll "github.com/darwinOrg/go-common/collection"
	dgctx "github.com/darwinOrg/go-common/context"
	dgerr "github.com/darwinOrg/go-common/enums/error"
	dgsys "github.com/darwinOrg/go-common/sys"
	dgcron "github.com/darwinOrg/go-cron"
	dglogger "github.com/darwinOrg/go-logger"
	"github.com/rolandhe/daog"
	"github.com/rolandhe/saber/gocc"
	"math"
	"time"
)

type CancelCallback func(ctx *dgctx.DgContext, tc *daog.TransContext) error

func InitTask(ctx *dgctx.DgContext, tc *daog.TransContext, taskType string, channel string, content string) (int64, error) {
	taskId, err := CommonTaskExtDao.InsertInitTask(tc, taskType, channel, content)
	if err != nil {
		dglogger.Errorf(ctx, "CommonTaskExtDao.InsertInitTask error: %v", err)
		return 0, err
	}
	return taskId, nil
}

func InitTaskWithBizId(ctx *dgctx.DgContext, tc *daog.TransContext, taskType, channel, content string, bizId int64) (int64, error) {
	taskId, err := CommonTaskExtDao.InsertInitTaskWithBizId(tc, taskType, channel, content, bizId)
	if err != nil {
		dglogger.Errorf(ctx, "CommonTaskExtDao.InsertInitTask error: %v", err)
		return 0, err
	}
	return taskId, nil
}

func InitDelayedTaskWithBizId(ctx *dgctx.DgContext, tc *daog.TransContext, taskType string, content string, bizId int64, delaySeconds int) (int64, error) {
	taskId, err := CommonTaskExtDao.InsertInitDelayedTaskWithBizId(tc, taskType, content, bizId, delaySeconds)
	if err != nil {
		dglogger.Errorf(ctx, "CommonTaskExtDao.InsertInitTask error: %v", err)
		return 0, err
	}
	return taskId, nil
}

func FetchTasks(ctx *dgctx.DgContext, tc *daog.TransContext, req *FetchTaskRequest) ([]*CommonTaskVo, error) {
	if req.PageSize == 0 {
		req.PageSize = 1
	}
	if req.LockMilli == 0 {
		req.LockMilli = 5000
	}
	if req.LockedBy == "" {
		req.LockedBy = dgsys.GetLocalLanIps()[0]
	}

	tasks, err := CommonTaskExtDao.LockTasksToProcessing(tc, req)
	if err != nil {
		dglogger.Errorf(ctx, "CommonTaskExtDao.FetchTasks error: %v", err)
		return nil, err
	}
	if len(tasks) == 0 {
		return []*CommonTaskVo{}, nil
	}
	return dgcoll.MapToList(tasks, ConvertTaskVo), nil
}

func EndAsSuccessByTc(ctx *dgctx.DgContext, tc *daog.TransContext, taskId int64) error {
	err := CommonTaskExtDao.EndAsSuccess(tc, taskId)
	if err != nil {
		dglogger.Errorf(ctx, "CommonTaskExtDao.EndAsSuccess error: %v", err)
	}
	return err
}

func EndAsSuccess(ctx *dgctx.DgContext, taskId int64) error {
	return daogext.Write(ctx, func(tc *daog.TransContext) error {
		err := CommonTaskExtDao.EndAsSuccess(tc, taskId)
		if err != nil {
			dglogger.Errorf(ctx, "CommonTaskExtDao.EndAsSuccess error: %v", err)
		}
		return err
	})
}

func EndTaskByError(ctx *dgctx.DgContext, task *CommonTaskVo, err error, maxProcessCount int32) {
	_ = daogext.Write(ctx, func(tc *daog.TransContext) error {
		if err.Error() != task.Reason && task.ProcessedCount > 0 {
			_ = CommonTaskExtDao.ResetProcessedCount(tc, task.Id)
			task.ProcessedCount = 0
		}

		if task.ProcessedCount < maxProcessCount {
			return EndAsFailByTc(ctx, tc, task.Id, err.Error())
		} else {
			return EndAsCanceledByTc(ctx, tc, task.Id, err.Error())
		}
	})
}

func EndTaskByErrorNotLimit(ctx *dgctx.DgContext, task *CommonTaskVo, err error) {
	EndTaskByError(ctx, task, err, math.MaxInt32)
}

func EndTaskByErrorTc(ctx *dgctx.DgContext, tc *daog.TransContext, task *CommonTaskVo, err error, maxProcessCount int32) error {
	if err.Error() != task.Reason && task.ProcessedCount > 0 {
		_ = CommonTaskExtDao.ResetProcessedCount(tc, task.Id)
		task.ProcessedCount = 0
	}

	if task.ProcessedCount < maxProcessCount {
		return EndAsFailByTc(ctx, tc, task.Id, err.Error())
	} else {
		return EndAsCanceledByTc(ctx, tc, task.Id, err.Error())
	}
}

func EndTaskByErrorWithCancelCallback(ctx *dgctx.DgContext, task *CommonTaskVo, err error, maxProcessCount int32, cancelCallback CancelCallback) {
	_ = daogext.Write(ctx, func(tc *daog.TransContext) error {
		return EndTaskByErrorTcWithCancelCallback(ctx, tc, task, err, maxProcessCount, cancelCallback)
	})
}

func EndTaskByErrorTcWithCancelCallback(ctx *dgctx.DgContext, tc *daog.TransContext, task *CommonTaskVo, err error, maxProcessCount int32, cancelCallback CancelCallback) error {
	if err.Error() != task.Reason && task.ProcessedCount > 0 {
		_ = CommonTaskExtDao.ResetProcessedCount(tc, task.Id)
		task.ProcessedCount = 0
	}

	if task.ProcessedCount < maxProcessCount {
		return EndAsFailByTc(ctx, tc, task.Id, err.Error())
	} else {
		ce := EndAsCanceledByTc(ctx, tc, task.Id, err.Error())
		if ce != nil {
			return ce
		}

		return cancelCallback(ctx, tc)
	}
}

func EndAsCanceledWithCancelCallback(ctx *dgctx.DgContext, taskId int64, reason string, cancelCallback CancelCallback) {
	_ = daogext.Write(ctx, func(tc *daog.TransContext) error {
		ce := EndAsCanceledByTc(ctx, tc, taskId, reason)
		if ce != nil {
			return ce
		}

		return cancelCallback(ctx, tc)
	})
}

func EndAsFailByTc(ctx *dgctx.DgContext, tc *daog.TransContext, taskId int64, reason string) error {
	err := CommonTaskExtDao.EndAsFail(tc, taskId, reason)
	if err != nil {
		dglogger.Errorf(ctx, "CommonTaskExtDao.EndAsFailByTc error: %v", err)
	}
	return err
}

func EndAsCanceledByTc(ctx *dgctx.DgContext, tc *daog.TransContext, taskId int64, reason string) error {
	err := CommonTaskExtDao.EndAsCanceled(tc, taskId, reason)
	if err != nil {
		dglogger.Errorf(ctx, "CommonTaskExtDao.EndAsCanceledByTc error: %v", err)
	}
	return err
}

func ReInit(ctx *dgctx.DgContext, tc *daog.TransContext, taskId int64) error {
	err := CommonTaskExtDao.ReInit(tc, taskId)
	if err != nil {
		dglogger.Errorf(ctx, "CommonTaskExtDao.ReInit error: %v", err)
	}
	return err
}

func GetByIdsAndLockedBy(ctx *dgctx.DgContext, tc *daog.TransContext, taskIds []int64, lockedBy string) ([]*CommonTaskVo, error) {
	tasks, err := CommonTaskExtDao.GetByIdsAndLockedBy(tc, taskIds, lockedBy)
	if err != nil {
		dglogger.Errorf(ctx, "CommonTaskExtDao.GetByIdsAndLockedBy error: %v", err)
		return nil, err
	}

	return dgcoll.MapToList(tasks, ConvertTaskVo), nil
}

func ReInitTimeoutProcessingTasks(ctx *dgctx.DgContext, taskType string, timeoutMinutes int) (int64, error) {
	return daogext.WriteWithResult(ctx, func(tc *daog.TransContext) (int64, error) {
		count, err := CommonTaskExtDao.ReInitTimeoutProcessingTasks(tc, taskType, timeoutMinutes)
		if err != nil {
			dglogger.Errorf(ctx, "CommonTaskExtDao.ReInitTimeoutProcessingTasks error: %v", err)
			return 0, err
		}
		if count > 0 {
			dglogger.Infof(ctx, "ReInitTimeoutProcessingTasks success, taskType: %s, count: %d", taskType, count)
		}

		return count, nil
	})
}

func CancelTaskById(ctx *dgctx.DgContext, tc *daog.TransContext, taskId int64, reason string) error {
	err := CommonTaskExtDao.CancelByIds(tc, []int64{taskId}, reason)
	if err != nil {
		dglogger.Errorf(ctx, "CommonTaskExtDao.CancelByIds error: %v", err)
		return err
	}

	return nil
}

func CancelTaskByIds(ctx *dgctx.DgContext, tc *daog.TransContext, taskIds []int64, reason string) error {
	err := CommonTaskExtDao.CancelByIds(tc, taskIds, reason)
	if err != nil {
		dglogger.Errorf(ctx, "CommonTaskExtDao.CancelByIds error: %v", err)
		return err
	}

	return nil
}

func FilterUnFinishedTaskIds(ctx *dgctx.DgContext, tc *daog.TransContext, taskIds []int64) ([]int64, error) {
	filteredTaskIds, err := CommonTaskExtDao.FilterUnFinishedTaskIds(tc, taskIds)
	if err != nil {
		dglogger.Errorf(ctx, "CommonTaskExtDao.FilterUnFinishedTaskIds error: %v", err)
		return nil, err
	}

	return filteredTaskIds, nil
}

func GetByTaskId(ctx *dgctx.DgContext, tc *daog.TransContext, taskId int64) (*CommonTaskVo, error) {
	task, err := CommonTaskDao.GetById(tc, taskId)
	if err != nil {
		dglogger.Errorf(ctx, "CommonTaskDao.GetById error: %v", err)
		return nil, err
	}
	if task == nil {
		return nil, nil
	}

	return ConvertTaskVo(task), nil
}

func ConvertTaskVo(task *CommonTask) *CommonTaskVo {
	return &CommonTaskVo{
		Id:             task.Id,
		TaskType:       task.Type,
		Channel:        task.Channel,
		Content:        task.Content.StringNilAsEmpty(),
		BizId:          task.BizId,
		ProcessedCount: task.ProcessedCount,
		Reason:         task.Reason.StringNilAsEmpty(),
	}
}

func RunSemaphoreTask(taskType string, limit uint, taskHandler func(ctx *dgctx.DgContext, task *CommonTaskVo)) {
	semaphore := gocc.NewDefaultSemaphore(limit)
	lockedBy := dgsys.GetLocalLanIps()[0]

	go func() {
		for {
			ctx := dgctx.SimpleDgContext()
			ctx.NotLogSQL = true
			task, err := daogext.WriteWithResult(ctx, func(tc *daog.TransContext) (*CommonTaskVo, error) {
				tasks, err := FetchTasks(ctx, tc, &FetchTaskRequest{
					TaskType: taskType,
					LockedBy: lockedBy,
				})
				if err != nil {
					return nil, err
				}
				if len(tasks) == 0 {
					return nil, nil
				}
				return tasks[0], nil
			})
			if err != nil || task == nil {
				time.Sleep(5 * time.Second)
				continue
			}

			acquired := dgcron.RunSemaphoreJob(ctx, taskType, semaphore, time.Minute, func(ctx *dgctx.DgContext) {
				taskHandler(ctx, task)
			})
			if !acquired {
				time.Sleep(5 * time.Second)
				EndTaskByErrorNotLimit(ctx, task, dgerr.TIME_OUT)
			}
		}
	}()
}

func DeleteOverdueTasks(ctx *dgctx.DgContext, overdueDays int) error {
	return daogext.Write(ctx, func(tc *daog.TransContext) error {
		err := CommonTaskExtDao.DeleteOverdueTasks(ctx, tc, overdueDays)
		if err != nil {
			dglogger.Errorf(ctx, "CommonTaskDao.DeleteOverdueTasks error: %v", err)
			return err
		}

		return nil
	})
}
